package net.pantolomin.nayo;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.AbstractSelectableChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.Iterator;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The selection loop
 */
@SuppressWarnings("synthetic-access")
class NioSelectionLoop implements Runnable {
	/** The logger */
	private final Logger logger = LoggerFactory.getLogger(NioSelectionLoop.class);

	/** The provider */
	private final NioProvider provider;

	/** The selector we'll be monitoring */
	private final Selector selector;

	/** The changes to apply on the selector */
	private final Queue<Runnable> changeRequests;

	/** Indicates if the loop must be stopped */
	private volatile boolean stopped;

	/**
	 * Create the selection loop
	 * 
	 * @param provider
	 *          the provider
	 * @throws IOException
	 *           if we failed to get a "selector"
	 */
	public NioSelectionLoop(NioProvider provider) throws IOException {
		this.provider = provider;
		this.selector = SelectorProvider.provider().openSelector();
		this.changeRequests = new ConcurrentLinkedQueue<Runnable>();
	}

	/**
	 * Stop all connections and shutdown the provider
	 */
	public void stop() {
		this.stopped = true;
		this.selector.wakeup();
	}

	public void run() {
		while (!this.stopped) {
			try {
				// Process any pending changes
				Runnable change;
				while ((change = this.changeRequests.poll()) != null) {
					change.run();
				}

				// Wait for an event on one of the registered channels
				this.selector.select(1000L);

				// Iterate over the set of keys for which events are available
				Iterator<SelectionKey> selectedKeys = this.selector.selectedKeys().iterator();
				while (selectedKeys.hasNext()) {
					SelectionKey key = selectedKeys.next();
					selectedKeys.remove();

					if (!key.isValid()) {
						continue;
					}

					// Check what event is available and deal with it
					try {
						if (key.isConnectable()) {
							finishConnection(key);
						} else if (key.isAcceptable()) {
							accept(key);
						} else {
							if (key.isWritable()) {
								// Now that we know that writing is possible, we
								// are not interested by the key
								key.interestOps(key.interestOps() ^ SelectionKey.OP_WRITE);
								write(key);
							}
							if (key.isReadable()) {
								// Since reading is done in another thread, we
								// need to remove the key
								// to prevent looping continuously
								key.interestOps(key.interestOps() ^ SelectionKey.OP_READ);
								read(key);
							}
						}
					} catch (Exception e) {
						this.logger.error("The handling of a selection key failed", e);
					}
				}
			} catch (Exception e) {
				this.logger.error("An error occurred in the selection process", e);
			}
		}
		try {
			this.selector.close();
		} catch (IOException e) {
			this.logger.warn("Failed to close the selector", e);
		}

		this.provider.closeAllConnections();
		this.changeRequests.clear();

		// In case we would want to restart
		this.stopped = false;
	}

	//
	// ************************************************************************
	// ************************************************************************
	// The operations
	// ************************************************************************
	// ************************************************************************

	/**
	 * Indicate that we are ready to accept connections on a server channel
	 * 
	 * @param serverChannel
	 *          the channel
	 * @param address
	 *          the remote address to bind on
	 */
	void acceptConnections(final ServerSocketChannel serverChannel, final InetSocketAddress address) {
		this.changeRequests.offer(new Runnable() {
			public void run() {
				try {
					serverChannel.register(selector, SelectionKey.OP_ACCEPT);
					try {
						serverChannel.socket().bind(address);
					} catch (IOException e) {
						// Unregister
						serverChannel.register(selector, 0);
						throw e;
					}
					provider.bound(serverChannel);
				} catch (IOException e) {
					provider.failedToBind(serverChannel, e);
				}
			}
		});
		this.selector.wakeup();
	}

	/**
	 * Indicate that we are waiting for a connection
	 * 
	 * @param socketChannel
	 *          the channel
	 * @param address
	 *          the remote address to connect to
	 */
	void makeConnection(final SocketChannel socketChannel, final InetSocketAddress address) {
		this.changeRequests.offer(new Runnable() {
			public void run() {
				try {
					socketChannel.register(selector, SelectionKey.OP_CONNECT);
					socketChannel.connect(address);
				} catch (IOException e) {
					provider.failedToConnect(socketChannel, e);
				}
			}
		});
		this.selector.wakeup();
	}

	/**
	 * Declare a timeout on a connection attempt
	 * 
	 * @param socketChannel
	 *          the channel
	 */
	void declareTimeout(final SocketChannel socketChannel) {
		this.changeRequests.offer(new Runnable() {
			public void run() {
				if (!socketChannel.isConnected()) {
					provider.connectionTimeout(socketChannel);
				}
			}
		});
		this.selector.wakeup();
	}

	/**
	 * Close a connection
	 * 
	 * @param socketChannel
	 *          the channel
	 * @param cleanly
	 *          indicates if the connection was closed cleanly
	 */
	void closeConnectionByRemote(final SocketChannel socketChannel, final boolean cleanly) {
		this.changeRequests.offer(new Runnable() {
			public void run() {
				provider.closeConnectionByRemote(socketChannel, cleanly);
			}
		});
		this.selector.wakeup();
	}

	/**
	 * Close a connection
	 * 
	 * @param socketChannel
	 *          the channel
	 */
	void closeChannel(final AbstractSelectableChannel socketChannel) {
		this.changeRequests.offer(new Runnable() {
			public void run() {
				provider.closeChannel(socketChannel);
			}
		});
		this.selector.wakeup();
	}

	/**
	 * Indicate that we are ready to read on a channel
	 * 
	 * @param selectionKey
	 *          the selection key
	 */
	void askForRead(final SelectionKey selectionKey) {
		this.changeRequests.offer(new Runnable() {
			public void run() {
				try {
					selectionKey.interestOps(selectionKey.interestOps() | SelectionKey.OP_READ);
				} catch (CancelledKeyException e) {
					// The key was canceled, the connection closed
					// => no more reading needed
				}
			}
		});
		this.selector.wakeup();
	}

	/**
	 * Indicate that we want to write on a channel
	 * 
	 * @param selectionKey
	 *          the selection key
	 */
	void askForWrite(final SelectionKey selectionKey) {
		this.changeRequests.offer(new Runnable() {
			public void run() {
				try {
					selectionKey.interestOps(selectionKey.interestOps() | SelectionKey.OP_WRITE);
				} catch (CancelledKeyException e) {
					// The key was canceled, the connection closed
					// => not possible to write
				}
			}
		});
		this.selector.wakeup();
	}

	/**
	 * Accept an incoming connection
	 * 
	 * @param key
	 *          the selection key
	 * @throws IOException
	 *           if an error occurred
	 */
	private void accept(SelectionKey key) throws IOException {
		// For an accept to be pending the channel must be a server socket
		// channel.
		ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();

		// Accept the connection and make it non-blocking
		SocketChannel socketChannel = serverSocketChannel.accept();
		socketChannel.configureBlocking(false);

		// Register the new SocketChannel with our Selector, indicating
		// we'd like to be notified when there's data waiting to be read
		SelectionKey selectionKey = socketChannel.register(this.selector, SelectionKey.OP_READ);
		this.provider.addSocketChannel(serverSocketChannel, socketChannel, selectionKey);
	}

	/**
	 * Finish the connection
	 * 
	 * @param key
	 *          the selection key
	 */
	private void finishConnection(SelectionKey key) {
		SocketChannel socketChannel = (SocketChannel) key.channel();

		// Finish the connection. If the connection operation failed
		// this will raise an IOException.
		try {
			socketChannel.finishConnect();
			this.provider.addSocketChannel(socketChannel, key);

			try {
				// We are now interested in reading
				key.interestOps(SelectionKey.OP_READ);
			} catch (CancelledKeyException e) {
				// The key was canceled, the connection closed
				// => no reading needed
			}
		} catch (IOException e) {
			// Cancel the channel's registration with our selector
			key.cancel();
			this.provider.failedToConnect(socketChannel, e);
		}
	}

	/**
	 * Read some incoming data
	 * 
	 * @param key
	 *          the selection key
	 * @throws IOException
	 *           if an error occurred
	 */
	private void read(SelectionKey key) throws IOException {
		this.provider.read((SocketChannel) key.channel());
	}

	/**
	 * Write some waiting data
	 * 
	 * @param key
	 *          the selection key
	 * @throws IOException
	 *           if an error occurred
	 */
	private void write(SelectionKey key) throws IOException {
		this.provider.write((SocketChannel) key.channel());
	}
}